package Autodesk;

import java.io.BufferedReader;
import java.io.FileReader;
import java.io.IOException;
import java.io.UnsupportedEncodingException;
import java.util.ArrayList;
import java.util.HashSet;
import java.util.List;
import java.util.Set;

import org.apache.commons.codec.DecoderException;
import org.apache.commons.codec.net.URLCodec;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.conf.Configured;
import org.apache.hadoop.filecache.DistributedCache;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.mapreduce.lib.output.MultipleOutputs;
import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;

public class WordFilter extends Configured implements Tool  {
	public Path prodPath;
	
	//---------------------------------------------------------------------
	// Mapper
	//---------------------------------------------------------------------

	public static class LogMapper extends Mapper<Object, Text, Text, IntWritable> {
		
		private final static IntWritable 		one 	= new IntWritable(1);
		private final static String[] ErrorCharacters 	= new String[]{"*","$","\""};
		private final static String ErrorSequence 		="title_sort";
		private Text word = new Text();
		
		
		// FIELDS ---------------------------------------------------------------------
		public Text getWord() {
			return word;
		}
		private Set<String> wordToExclude=new HashSet<String>();		
		public void setWord(Text inputWord){
			word=inputWord;
		}
		public void setWordsToExclude(HashSet<String> list){
			wordToExclude=list;
			System.out.println("LIST"+wordToExclude.toString());
		}
		
		// SETUP BEFORE MAPPING---------------------------------------------------------------------
		/**
		 * Method that get the words that should be excluded from the processing
		 * @param Context
		 * @return The proccess is ready to start the mapping
		 * @author leandro.mora
		 */
		public void setup(Context context){
			Configuration conf=context.getConfiguration();
			 if (conf.getBoolean("arch.patterns", false)) {
				  Path[] patternsFiles = new Path[0];
				  try{
			           patternsFiles = DistributedCache.getLocalCacheFiles(conf);
				  }catch(IOException ioe){
			           System.err.println("Caught exception while getting cached files: ");
				  }
				  for (Path patternsFile : patternsFiles) {
					  	getStaticList(patternsFile,wordToExclude); 
				  }
				 
			 }
		}
		
		private void getStaticList(Path patternFile,Set<String> staticList){
			try{
				BufferedReader fis=new BufferedReader(new FileReader(patternFile.toString()));	
				String pattern=null;
				
				while((pattern =fis.readLine())!=null){
						staticList.add(pattern.toString());
				}
				fis.close();
			}catch(IOException ioe){
		         System.err.println("Caught exception while parsing the cached file '");
			}
		}		

		//---------------------------------------------------------------------

		/**
		 * Map of the word filter
		 * @param Context
		 * @return key(DATE,LANGUAGE,CHANNEL,PRODUCT,COUNTRY, WORD) Value(1)
		 * @author leandro.mora
		 */
		public void map(Object key, Text value, Context context)
			throws IOException, InterruptedException {
			StringBuffer bf = new StringBuffer();
	
			String record[] = value.toString().toLowerCase().split("\t");
			try{
				bf.append(record[0]);// Date.
				bf.append("\t");
				bf.append(record[4]);// Language
				bf.append("\t");
				bf.append(record[1]);// Channel
				bf.append("\t");
				bf.append(record[3]);// Product
				bf.append("\t");
				bf.append(record[8]);// Country
				bf.append("\t");
				ArrayList<String> listWords = getWordsList(record[6]);								
				for(int i=0;i<listWords.size();i++){
					if(!(wordToExclude.contains(listWords.get(i))) && !listWords.get(i).isEmpty() && !(listWords.get(i).contains(ErrorSequence))){
						word.set(bf.toString()+listWords.get(i));
						context.write(word, one);
					}
				}
			}catch(Exception e){
				System.out.println("Erroor Reg: "+value);
				e.printStackTrace();				
			}
		}
		
		//---------------------------------------------------------------------
		private String excludedErrorCharacters(String word){
			for(int i=0;i<ErrorCharacters.length;i++){
				if(word.endsWith(ErrorCharacters[i])){
					word=word.substring(0, word.length()-1);
					break;
				}
			}
			for(int i=0;i<ErrorCharacters.length;i++){
				if(word.startsWith(ErrorCharacters[i])){
					word=word.substring(1, word.length());
					break;
				}
			}
			return word;
		}
		/**
		 * Method that extracts the words from a query passed as parameter
		 * @param query
		 * @return The list of words from that query
		 * @throws UnsupportedEncodingException
		 * @throws DecoderException
		 * @author leandro.mora
		 */
		public ArrayList<String> getWordsList(String query)
			throws UnsupportedEncodingException, DecoderException {
			byte[] buf=query.getBytes("UTF-8");
			byte[] buf2=URLCodec.decodeUrl(buf);
			String queryParsed = new String(buf2);	
			ArrayList<String> listWords=new ArrayList<String>();
			queryParsed=TotalQueryFilter.eliminateTabsQuery(queryParsed);
			queryParsed=queryParsed.replaceAll("\\s+", " ");
			String[] words=queryParsed.toString().split(" ");
	
			for(int f=0;f<words.length;f++){
					String finalWord=this.excludedErrorCharacters(words[f]);
					listWords.add(finalWord);
			}
				return listWords;
			}
		}
	//---------------------------------------------------------------------
	// Reducer
	//---------------------------------------------------------------------

	public static class LogReducer extends
			Reducer<Text, IntWritable, Text, IntWritable> {
		
		private IntWritable result = new IntWritable(1);
		private MultipleOutputs<Text,IntWritable> mos;
		 
		 public MultipleOutputs<Text, IntWritable> getMos() {
			return mos;
		}

		public void setMos(MultipleOutputs<Text, IntWritable> mos) {
			this.mos = mos;
		}

		public void setup(Context context) {
			 mos = new MultipleOutputs<Text,IntWritable>(context);
		 }
		
		public void reduce(Text key, Iterable<IntWritable> values,
				Context context) throws IOException, InterruptedException {
			
			int sum = 0;
			for (IntWritable val : values) {
				sum += val.get();
			}
			if(sum!=0){
				result.set(sum);
			}
			//context.write(key, result);
			String[] keySplitted=key.toString().split("\t",2);
			mos.write(key, result, keySplitted[0]+"_WordReport");
		}
		 public void cleanup(Context context) throws IOException, InterruptedException {
			 mos.close();
		 }
	}
	
	//---------------------------------------------------------------------
	// Configuration
	//---------------------------------------------------------------------
	
	public int run(String[] args) throws Exception { 
		Configuration conf = getConf();
		
		Job job = new Job(conf, "ParseLogs");
		job.setJarByClass(Autodesk.TotalVolumeFilter.class);
	
		job.setMapperClass(LogMapper.class);
		job.setReducerClass(LogReducer.class);
		
		job.setOutputKeyClass(Text.class);
		job.setOutputValueClass(IntWritable.class);
	
		List<String> other_args = new ArrayList<String>();
		for (int i = 0; i < args.length; ++i) {
			if ("-arch".equals(args[i])) {
				prodPath=new Path(args[++i]);
				DistributedCache.addCacheFile(prodPath.toUri(),
						job.getConfiguration());
				job.getConfiguration().setBoolean("arch.patterns",
						true);
				job.getConfiguration().set("arch.path", prodPath.getName());
			} else{
				other_args.add(args[i]);
			}
		}
	    FileInputFormat.setInputPaths(job, new Path(other_args.get(0)));
	    FileOutputFormat.setOutputPath(job, new Path(other_args.get(1)));
	    MultipleOutputs.addNamedOutput(job,"WordReport", TextOutputFormat.class, Text.class, IntWritable.class);
	
		System.exit(job.waitForCompletion(true)?0:1);
		
		return 0;	
	}
	
	//---------------------------------------------------------------------
	// Runner
	//---------------------------------------------------------------------
		
	public static void main(String[] args) throws Exception {
		int res = ToolRunner.run(new Configuration(), new WordFilter(), args);
		System.exit(res);
	}

}
